package cn.yanceysong.workcount;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class WordCountDemoMain {
    static Logger logger = LoggerFactory.getLogger(WordCountDemoMain.class);

    public static void main(String[] args) throws Exception {
        String host = "172.17.43.148";
        int port = 6666;

        // set up the execution environment
       final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 创建配置对象
        Configuration config = new Configuration();
        // 设置 Web UI 端口为 9000
        config.set(RestOptions.PORT, 8050);

        // source
        SocketTextStreamFunction socketTextStreamFunction = new SocketTextStreamFunction(host, port, "\n", 0);
        DataStream<String> source = env.addSource(socketTextStreamFunction, "Source")
                .name("Source");

        // transform
        DataStream<Tuple2<String, Integer>> countsDataStream =
                // split up the lines in pairs (2-tuples) containing: (word,1)
                source.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, collector) -> {
                            // normalize and split the line
                            String[] tokens = value.toLowerCase().split("\\W+");

                            // emit the pairs
                            for (String token : tokens) {
                                if (!token.isEmpty()) {
                                    collector.collect(new Tuple2<>(token, 1));
                                }
                            }
                        })
                        // 帮助jdk推断类型
                        .returns(Types.TUPLE(Types.STRING, Types.INT))
                        // group by the tuple field "0" and sum up tuple field "1"
                        .keyBy(value -> value.f0)
                        .sum(1)
                        .name("Transform");

        // sink = log
        // 这里为了方便展示效果，将结果直接输出到 log
        countsDataStream.addSink(new WordCountSink()).name("Sink");
        countsDataStream.print();
        // execute program
        env.execute("WordCount from socket.");
    }
}
